package mongo.db.clients.firsts;

import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.Set;

import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.mongodb.MapReduceCommand;
import com.mongodb.MapReduceCommand.OutputType;
import com.mongodb.MapReduceOutput;
import com.mongodb.MongoClient;
import com.mongodb.ServerAddress;

public class NetflowMapReduce {

	/**
	 * @param args
	 * @throws UnknownHostException 
	 * @throws InterruptedException 
	 */
	public static void main(String[] args) throws UnknownHostException, InterruptedException {

//		MongoClient client = new MongoClient("localhost", 27011);
		List<ServerAddress> seeds = new ArrayList<ServerAddress>();
		
		seeds.add(new ServerAddress("SS59-MAA", 27021));
		seeds.add(new ServerAddress("SS59-MAA", 27022));
		
		MongoClient client = new MongoClient(seeds);
		
		DB db = client.getDB("netflow");
		DBCollection collection = db.getCollection("flows");
		
		System.out.println("count : " + collection.count());
		DBCursor iter = collection.find();
		
/*		while ( iter.hasNext()) {
			System.out.println(iter.next());
		}*/
		

		Random rn = new Random();
		
		String map = "function() { emit ( this.routerip, this.octets ); };"; 
		String reduce = "function(ip, octet) { return Array.sum(octet); };";
		
		MapReduceCommand mrc = new MapReduceCommand(collection, map, reduce, "tots", OutputType.MERGE , null);
		
		
		MapReduceOutput mapout = collection.mapReduce(mrc);
		System.out.println("map reduce output : ");
		for ( DBObject o : mapout.results()) {
			System.out.println(o.toString());
		}
		
		Set<String> all = db.getCollectionNames();
		System.out.println("collections of reduce : " + all);
		
	}

}
